# -*- coding: utf-8 -*-
# This code written by ChanGOGOGO
import base64
import calendar
import datetime
import decimal
import hashlib
import json
import logging
import math
import os
import smtplib
import socket
import time
import zipfile
import requests
import xlsxwriter
import yaml
from elasticsearch import Elasticsearch, ConnectionError, TransportError
from email.header import Header
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from functools import wraps
from logging.handlers import RotatingFileHandler
from typing import Union


def setup_logging(log_level: str = "INFO", write_log: bool = True):
    """
    日志格式化打印
    :param log_level: 打印高于等于设定级别的日志。DEBUG < INFO < WARNING < ERROR < CRITICAL
    :param write_log: 是否需要将日志写入到文件
    :return: None
    """
    # 创建日志器
    logger = logging.getLogger()
    logger.setLevel(log_level)  # 设置日志级别为DEBUG
    # 设置日志格式
    formatter = logging.Formatter(
        "%(asctime)s [%(levelname)s]：%(message)s", datefmt="%Y-%m-%d %H:%M:%S"
    )
    # 创建一个流处理器，输出到控制台
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter)
    logger.addHandler(console_handler)
    # 根据参数决定是否也将日志写入文件
    if write_log:
        # 确保日志目录存在
        if not os.path.exists("logs"):
            os.makedirs("logs")
        # 创建一个循环文件处理器，自动根据大小切割
        file_handler = RotatingFileHandler(
            "logs/info.log", maxBytes=100 * 1024 * 1024, backupCount=5
        )
        file_handler.setFormatter(formatter)
        logger.addHandler(file_handler)


setup_logging()


def read_config(config_path):
    """读取config.yaml配置文件"""
    with open(config_path, encoding="utf-8") as f:
        config = yaml.safe_load(f)
    return config


def get_local_ip():
    """获取本机内网IP"""
    try:
        # 连接到 Google 的公共 DNS (8.8.8.8) 端口 80, 仅用于获取本机局域网IP
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.settimeout(0)
        s.connect(("8.8.8.8", 80))
        ip = s.getsockname()[0]
        s.close()
        return ip
    except Exception as e:
        return str(e)


local_ip = get_local_ip()
if local_ip.startswith("192.168"):
    env = "pet"
else:
    env = "pls"
# 指定环境配置文件
env_config = f"./config/{env}.yaml"
server_config = read_config(env_config)
baidu_minio_address = server_config.get("baidu_minio").get("address")
baidu_minio_access_key = server_config.get("baidu_minio").get("access_key")
baidu_minio_secret_key = server_config.get("baidu_minio").get("secret_key")
old_minio_address = server_config.get("old_minio").get("address")
old_minio_access_key = server_config.get("old_minio").get("access_key")
old_minio_secret_key = server_config.get("old_minio").get("secret_key")
server_ip = server_config.get("server_ip")
server_port = server_config.get("server_port")
excel_save_dir = server_config.get("excel_save_dir")
db_ip = server_config.get("mysql").get("ip")
db_port = server_config.get("mysql").get("port")
db_username = server_config.get("mysql").get("username")
db_password = server_config.get("mysql").get("password")
es_ip = server_config.get("elasticsearch").get("ip")
es_port = server_config.get("elasticsearch").get("port")
push_url = server_config.get("lemon_push_url")
per_excel_data_num = server_config.get("per_excel_data_num")
install_package_dir = server_config.get("install_package_dir")
lastst_version = server_config.get("latest_client_version")
first_second_last_part = server_config.get("first_second_last_part")

# 会话状态
session_status_infos = {
    1: "未开始",
    3: "准备队列等待中",
    5: "等待中",
    8: "延时",
    9: "进行队列等待中",
    10: "进行中",
    20: "已暂停",
    21: "已暂停",
    30: "已完成",
    40: "失败",
    50: "已关闭",
}

# 呼出记录状态
callout_connect_status_infos = {
    100: "已接通",
    0: "尚未拨打",
    1: "未接通-关机",
    2: "未接通-通话中",
    3: "未接通-不在服务区",
    4: "未接通-欠费",
    5: "未接通-无人接听",
    6: "未接通-暂时无法接通",
    7: "未接通-停机",
    8: "未接通-空号",
    9: "未接通-电路正忙",
    10: "未接通-用户正忙",
    11: "未接通-其他原因",
    12: "未接通-请勿越权使用",
    17: "未接通-系统正忙",
    18: "未接通-暂停服务",
    19: "未接通-用户挂断",
    80: "未拨打-API干预取消拨打",
    81: "未拨打-任务终止",
    82: "未拨打-任务到达结束时间",
    83: "未拨打-手动停止拨打",
    84: "未拨打-呼叫限制",
    97: "未拨打-黑名单",
    98: "未拨打-外呼异常",
}
# 呼出任务状态
task_status = {
    0: "未开始",  # 创建任务，尚未导入号码
    1: "未开始",  # 导入号码完成
    5: "队列中",
    10: "进行中",  # 包含：进行中、空闲、非营业时段
    20: "手动暂停",
    21: "异常暂停",
    25: "已完成",  # 从界面点了任务结束, 作为中间状态，事实上是任务终止中
    30: "已完成",
    40: "已过期",
    50: "被删除",  # 任务被删除掉了
}

# 挂断者
dialog_signoff_trigger_infos = {1: "用户", 2: "机器人", 3: "未接通"}

# 播音模式
play_mode_infos = {
    "playMixTts": "混合模式",
    "playPureTts": "仅语音合成",
    "playAudio": "纯录音",
}


def get_col_name(col_index):
    """根据excel列序号，获取列对应的字母名。例如：1对应A，3对应C，29对应AC"""
    result = []
    while col_index > 0:
        col_index, remainder = divmod(col_index - 1, 26)
        result.append(chr(65 + remainder))
    return "".join(reversed(result))


def write_excel(
    sheet_infos: list, save_dir: str, file_name="导出数据", need_timestamp=True
):
    """
    生成excel数据文件
    :param sheet_infos: 每个工作表的信息。组成元素为[sheet页名称, 第一行标题list, 第二行及之后的数据(嵌套list)]
    example = [
        [
            '人员信息',  # sheet名称
            ['姓名', '年级'],  # 第一行数据，也就是表格标题
            [
                ['小明', 6],  # 第2-N行的数据
                ['小红', 7]
            ]
        ],
        [
            '成绩信息',
            ['姓名', '科目', '成绩'],
            [
                ['小明', '语文', '99'],
                ['小明', '数学', 60]
            ]
        ]
    ]
    :param save_dir: excel文件保存目录
    :param file_name: excel文件名，不带后缀（后缀自动添加.xlsx）
    :param need_timestamp: excel文件名后是否需要加上时间戳
    :return: excel_name
    """
    start_time = time.time()
    # 文件名是否需要时间戳
    now_time = time.strftime("%Y%m%d%H%M%S")
    if need_timestamp:
        excel_name = f"{file_name}_{now_time}.xlsx"
    else:
        excel_name = f"{file_name}.xlsx"
    save_path = os.path.join(save_dir, excel_name)  # excel文件保存路径
    work_book = xlsxwriter.Workbook(save_path, {"constant_memory": True})
    title_style = work_book.add_format(
        {"bold": True, "align": "center", "valign": "vcenter", "fg_color": "00C957"}
    )
    row_style = work_book.add_format({"text_wrap": True, "valign": "vcenter"})
    for sheet_info in sheet_infos:
        sheet_name, sheet_title, sheet_data = sheet_info
        sheet = work_book.add_worksheet(sheet_name)
        sheet.write_row("A1", sheet_title, cell_format=title_style)
        col_index = 1
        for title in sheet_title:
            col_name = get_col_name(col_index)
            title_len = len(title)
            col_width = title_len * 2.5
            sheet.set_column(f"{col_name}:{col_name}", col_width)
            col_index += 1
        sheet.freeze_panes(1, 0)  # 冻结首行
        row_num = 1
        for row_data in sheet_data:
            row_num += 1
            sheet.write_row(f"A{row_num}", data=row_data, cell_format=row_style)
    work_book.close()
    end_time = time.time()
    take_time = end_time - start_time
    logging.info(f"生成「{excel_name}」共计耗时{take_time:.2f}秒。")
    return excel_name


def write_excel2(
    sheet_info: list, save_dir: str, file_name="导出数据", need_timestamp=True
):
    """
    针对大模型质检生成excel数据文件
    :param sheet_info: 每个工作表的信息。组成元素为[sheet页名称, 第一行标题list, 第二行及之后的数据(嵌套list)]
    example =
        [
            '人员信息',  # sheet名称
            ['姓名', '年级'],  # 第一行数据，也就是表格标题
            [
                ['小明', 6],  # 第2-N行的数据
                ['小红', 7]
            ]
        ]
    :param save_dir: excel文件保存目录
    :param file_name: excel文件名，不带后缀（后缀自动添加.xlsx）
    :param need_timestamp: excel文件名后是否需要加上时间戳
    :return: None
    """
    start_time = time.time()
    # 文件名是否需要时间戳
    now_time = time.strftime("%Y%m%d%H%M%S")
    if need_timestamp:
        full_file_name = f"{file_name}_{now_time}.xlsx"
    else:
        full_file_name = f"{file_name}.xlsx"
    save_path = os.path.join(save_dir, full_file_name)  # excel文件保存路径
    work_book = xlsxwriter.Workbook(save_path, {"constant_memory": True})
    title_style = work_book.add_format(
        {"bold": True, "align": "center", "valign": "vcenter", "fg_color": "00C957"}
    )
    row_style = work_book.add_format({"text_wrap": True, "valign": "vcenter"})
    sheet_name, sheet_title, sheet_data = sheet_info
    sheet = work_book.add_worksheet(sheet_name)
    sheet.write_row("A1", sheet_title, cell_format=title_style)
    sheet.set_column("A:A", 11)
    sheet.set_column("B:B", 8)
    sheet.set_column("C:C", 11)
    sheet.set_column("D:D", 11)
    sheet.set_column("E:E", 22)
    sheet.set_column("F:F", 25)
    sheet.set_column("G:G", 15)
    sheet.set_column("H:H", 15)
    sheet.set_column("I:I", 15)
    sheet.set_column("J:J", 15)
    sheet.set_column("K:K", 15)
    sheet.set_column("L:L", 16)
    sheet.set_column("M:M", 50)
    sheet.set_column("N:N", 16)
    sheet.set_column("O:O", 40)
    sheet.set_column("P:P", 16)
    sheet.set_column("Q:Q", 40)
    sheet.set_column("R:R", 12)
    sheet.set_column("S:S", 40)
    sheet.set_column("T:T", 18)
    sheet.set_column("U:U", 40)
    sheet.freeze_panes(1, 0)  # 冻结首行
    row_num = 1
    for row_data in sheet_data:
        row_num += 1
        sheet.write_row(f"A{row_num}", data=row_data, cell_format=row_style)
    work_book.close()
    end_time = time.time()
    take_time = end_time - start_time
    logging.info(f"将数据写入【{full_file_name}】共计耗时{take_time:.2f}秒。")
    return full_file_name


def ms2min(ms_value: int):
    """
    毫秒时长转换成分钟时长，向上取整
    :param ms_value: 毫秒时长
    :return: 分钟时长
    """
    min_value = math.ceil(ms_value / 60000)
    return min_value


def check_str_is_num(str_object: str) -> bool:
    """
    判断字符串是不是纯数字
    001 ——> True
    福利告知3 ——> False
    正式会员 ——> False
    :param str_object: 字符串对象
    :return: Bool值
    """
    return str_object.isdecimal()


def prefix_contain_str(str_object: str, prefix_list: list):
    """
    检查字符串是否以列表中的元素开头
    :param str_object: 字符串数据
    :param prefix_list: 字符串前缀组成的列表
    :return: 字符串以列表中的元素为前缀，返回True；否则，返回False
    """
    check_result = True
    for prefix in prefix_list:
        if str_object.find(str(prefix)) != -1:
            check_result = True
            return check_result
        else:
            check_result = False
    return check_result


def prefix_start_with(str_object: str, prefix_list: list):
    """
    检查字符串是否以给定的前缀列表中的任何一个前缀开始。

    :param str_object: 待检查的字符串。
    :param prefix_list: 包含多个前缀的列表。
    :return: 如果字符串以列表中的任何一个前缀开始，则返回True；否则返回False。
    """
    # 初始化检查结果为True，假设字符串以某个前缀开始
    check_result = True

    # 遍历前缀列表，检查字符串是否以任何一个前缀开始
    for prefix in prefix_list:
        # 如果找到匹配的前缀，设置检查结果为True并立即返回
        if str_object.startswith(str(prefix)):
            check_result = True
            return check_result
        else:
            # 如果当前前缀不匹配，设置检查结果为False，但继续检查其他前缀
            check_result = False

    # 如果所有前缀都不匹配，返回False
    return check_result


def is_contain_str(str_object: str, contain_str_list: list):
    """验证字符串内容是否包含字符串列表中的某个子字符串"""
    check_result = True
    for contain_str in contain_str_list:
        if str_object.find(str(contain_str)) != -1:
            check_result = True
            return check_result
        else:
            check_result = False
    return check_result


def are_contain_str(str_objects: list, contain_str_list):
    """查询一组字符串中每一个字符串是否包含某一组关键字中其中任意一个关键字"""
    current_check_result = False
    for str_object in str_objects:
        check_result = is_contain_str(str_object, contain_str_list)
        if check_result:
            return check_result
    return current_check_result


def data2beautify(data: Union[dict, list]):
    """
    字典或列表类型的数据，进行美化输出
    :param data: 字典或列表类型数据
    :return: 格式化数据
    """
    if not isinstance(data, (dict, list)):
        raise TypeError("Input data must be a dictionary or list")

    try:
        formatted_data = json.dumps(data, indent=4, ensure_ascii=False)
    except (TypeError, ValueError) as e:
        # 记录异常信息，便于调试
        print(f"Error formatting data: {e}")
        raise

    return formatted_data


def get_time_desc(total_seconds):
    """将秒数转换为X分、X小时等更加清晰易懂的描述"""
    total_seconds = int(total_seconds)
    if total_seconds < 1:
        time_desc = f"不到1秒"
    elif 1 <= total_seconds < 60:
        time_desc = f"{total_seconds}秒"
    elif 60 <= total_seconds < 3600:
        minutes, remainder_seconds = divmod(total_seconds, 60)
        if remainder_seconds == 0:
            time_desc = f"{minutes}分钟"
        else:
            time_desc = f"{minutes}分{remainder_seconds}秒"
    else:
        hours, remainder_seconds1 = divmod(total_seconds, 3600)
        minutes, remainder_seconds2 = divmod(remainder_seconds1, 60)
        if minutes == 0:
            if remainder_seconds2 == 0:
                time_desc = f"{hours}小时"
            else:
                time_desc = f"{hours}小时{remainder_seconds2}秒"
        else:
            if remainder_seconds2 == 0:
                time_desc = f"{hours}小时{minutes}分钟"
            else:
                time_desc = f"{hours}小时{minutes}分{remainder_seconds2}秒"
    return time_desc


def get_nearby_time(target_time, days: int = -1):
    """
    获取指定时间前或者后N天的相同时间点
    :param target_time: 指定时间
    :param days: 几天前用负数，几天后用正数
    :return: 几天前或几天后的时间
    """
    start_time = datetime.datetime.strptime(target_time, "%Y-%m-%d %H:%M:%S")
    pass_time = datetime.timedelta(days=days)
    end_time = str(start_time + pass_time)
    return end_time


def get_nearby_day(target_date, days: int = -1):
    """
    获取指定日期前后几天的日期
    :param target_date: 指定日期
    :param days: 几天前用负数，几天后用整数
    :return: 几天前或几天后的日期
    """
    format_target_date = datetime.datetime.strptime(target_date, "%Y-%m-%d")
    pass_date = datetime.timedelta(days=days)
    end_time = str(format_target_date + pass_date)
    end_date = end_time[0:10]
    # logging.info(f'{target_date}的前后{days}天是：{end_date}')
    return end_date


def get_current_month():
    """获取当前月份"""
    current_month = time.strftime("%Y%m", time.localtime())
    return current_month


def get_today_date():
    today = time.strftime("%Y-%m-%d", time.localtime())
    return today


def get_date_from_time(target_time):
    """给定一个时间，获取日期"""
    # 将字符串时间转换为datetime对象
    datetime_obj = datetime.datetime.strptime(target_time, "%Y-%m-%d %H:%M:%S")
    # 从datetime对象中提取日期部分，并格式化为字符串
    date = datetime_obj.strftime("%Y-%m-%d")
    return date


def get_weekday(date_str):
    """获取日期对应的是周几，日期格式为YYYY-mm-dd"""
    # 将字符串日期转换为datetime对象
    date_obj = datetime.datetime.strptime(date_str, "%Y-%m-%d")
    # 获取星期几，返回的是一个整数，其中0是星期一，6是星期日
    week_day = date_obj.weekday()
    # 将整数转换为星期几的名称，以星期一作为一周的第一天
    weekdays = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"]
    return weekdays[week_day]


def get_start_end_time(target_date=None):
    """获取今天的开始结束时间"""
    if target_date is None:
        target_date = get_today_date()
    start_time = f"{target_date} 00:00:00"
    end_time = f"{target_date} 23:59:59"
    return start_time, end_time


def get_date_month(target_date):
    # 将字符串转换为日期对象
    date_obj = datetime.datetime.strptime(target_date, "%Y-%m-%d")
    # 获取年份和月份并拼接成想要的格式
    result = date_obj.strftime("%Y%m")
    return result


def in_the_same_month(first_time, second_time):
    """判断2个时间是否在同一个月"""
    first_month = first_time[0:7]
    second_month = second_time[0:7]
    if first_month == second_month:
        return True
    else:
        return False


def get_month_first_last_day(target_time):
    """获得指定时间所在月份的第一天和最后一天"""
    year = int(target_time[0:4])
    month = int(target_time[5:7])
    # 获取当前月的第一天的星期和当月总天数
    _, month_count_day = calendar.monthrange(year, month)
    # 获取当前月份第一天
    first_day = datetime.date(year, month, day=1)
    # 获取当前月份最后一天
    last_day = datetime.date(year, month, day=month_count_day)
    # 返回第一天和最后一天
    return first_day, last_day


def get_month_time_scale(date_month="202403"):
    """
    获取指定月份的第一天00:00:00和最后一天23:59:59
    :param date_month:
    :return:
    """
    year = int(date_month[0:4])
    month = int(date_month[4:])

    # 获取本月第一天
    this_month_start = datetime.datetime(year, month, 1)
    this_month_start = datetime.datetime.strftime(this_month_start, "%Y-%m-%d %H:%M:%S")
    # 获取本月最后一天
    last_day_of_month = calendar.monthrange(year, month)[1]
    this_month_end = datetime.datetime(year, month, last_day_of_month, 23, 59, 59)
    this_month_end = datetime.datetime.strftime(this_month_end, "%Y-%m-%d %H:%M:%S")
    return this_month_start, this_month_end


def get_file_size(file_path):
    """获取文件大小，单位：兆"""
    size = os.path.getsize(file_path)
    size_mb = real_round(size / 1024 / 1024, 2)
    return size_mb


def get_files_size(file_paths):
    files_size = 0
    for file_path in file_paths:
        files_size += get_file_size(file_path)
    return files_size


def remove_file(file_path):
    """删除文件"""
    if os.path.exists(file_path):
        os.remove(file_path)


def remove_expired_files(exec_dir, expired_days: int = -7):
    """删除expired_days天前或者后的文件"""
    file_names = os.listdir(exec_dir)
    now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
    expired_time = get_nearby_time(now_time, expired_days)
    for file_name in file_names:
        file_path = os.path.join(exec_dir, file_name)
        create_time = os.path.getctime(file_path)
        create_time_format = time.strftime(
            "%Y-%m-%d %H:%M:%S", time.localtime(create_time)
        )
        if create_time_format <= expired_time:
            os.remove(file_path)


def wecom_push_msg(qw_push_url, msg):
    """向企微群推送消息"""
    data = {"msgtype": "markdown", "markdown": {"content": msg}}
    logging.info(f"向企微群推送消息接口发起请求：\n{data2beautify(data)}")
    res = requests.post(qw_push_url, data=json.dumps(data)).json()
    logging.info(f"收到企微群推送消息接口响应：\n{data2beautify(res)}")
    return res


def post_json(url: str, data: dict = None, log_switch: bool = True, **kwargs):
    """
    发送post请求，请求内容为json数据
    :param url: 请求url
    :param data: 请求体
    :param log_switch: 日志开关
    :return: 接口响应
    """
    headers = kwargs.get("headers")
    if headers:
        ...
    else:
        headers = {"Content-Type": "application/json"}
    timeout = kwargs.get("timeout")
    res = requests.post(
        url, data=json.dumps(data), headers=headers, timeout=timeout
    ).json()
    if log_switch:
        logging.info(f"[请求消息]向{url}接口发起POST请求：\n{data2beautify(data)}")
        logging.info(f"[响应消息]从{url}接口收到响应：\n{data2beautify(res)}")
    return res


def send_email_with_attch(
    send_server_add,
    send_server_port,
    sender: str,
    sender_password: str,
    receivers: list,
    email_title,
    main_body,
    attachment_paths: list,
):
    """"""
    # 创建一个带附件的对象
    message = MIMEMultipart()
    from_name_base64 = base64.b64encode(bytes("追一科技", "utf-8"))
    from_name_str = str(from_name_base64, "utf-8")
    from_name_str = '"=?utf-8?B?' + from_name_str + '=?=" <' + sender + ">"
    message["FROM"] = Header(from_name_str)
    message["Subject"] = Header(email_title, "utf-8")
    # 邮件正文内容
    message.attach(MIMEText(main_body, "plain", "utf-8"))
    # 附件
    for attachment_path in attachment_paths:
        attachment_name = attachment_path.split(os.sep)[-1]
        with open(attachment_path, "rb") as file:
            file_stream = file.read()
        attachment = MIMEText(file_stream, "base64", "utf-8")
        attachment.add_header(
            "Content-Disposition",
            "attachment",
            filename=Header(f"{attachment_name}", "utf-8").encode(),
        )
        message.attach(attachment)
    # 发送邮件
    smtp = smtplib.SMTP_SSL(send_server_add, send_server_port, timeout=3000)
    smtp.login(sender, sender_password)
    smtp.sendmail(sender, receivers, message.as_string())
    smtp.close()


def real_round(float_num: float, ndigits: int = 2):
    """
    浮点数四舍五入(避开round函数的坑)
    :param float_num: 浮点数
    :param ndigits: 小数位精确度
    :return: 四舍五入后的浮点数
    """
    if ndigits == 0:
        num_template = "0"
    else:
        num_template = "0." + ("0" * ndigits)
    decimal.getcontext().rounding = "ROUND_HALF_UP"
    precision_num = decimal.Decimal(str(float_num)).quantize(
        decimal.Decimal(num_template)
    )  # 需要把数字转换成字符串才能处理
    return float(precision_num)


def calc_percent(numerator, denominator, reserved_bits=2):
    """保留两位小数，四舍五入"""
    try:
        percent = (numerator / denominator) * 100
        return f"{percent:.{reserved_bits}f}%"
    except ZeroDivisionError:
        return "0.00%"


def package_file(file_paths, zip_name):
    """将多个文件打压缩包"""
    with zipfile.ZipFile(zip_name, "w") as zip_file:
        for file_path in file_paths:
            file_name = file_path.split(os.sep)[-1]
            zip_file.write(file_path, arcname=file_name)


def file_grouping(file_paths, files_size_limit=50):
    """对文件进行分组，每个分组文件大小不超过files_size_limit"""
    file_groups = []
    for file_path in file_paths:
        file_size = get_file_size(file_path)
        for file_group in file_groups:
            files_size = get_files_size(file_group)
            if files_size + file_size <= files_size_limit:
                file_group.append(file_path)
                break
        else:
            file_groups.append([file_path])
    return file_groups


def get_chinese_date(target_time):
    # 使用strptime将字符串转换为datetime对象
    date_obj = datetime.datetime.strptime(target_time, "%Y-%m-%d %H:%M:%S")
    # 使用strftime将datetime对象格式化为所需的格式
    chinese_date = date_obj.strftime("%Y年%m月%d日")
    return chinese_date


def query_es(es_index, body, fields, ip=es_ip, port=es_port, scroll="5m"):
    es = None
    scroll_id = None
    start_time = time.time()
    data = list()

    try:
        es = Elasticsearch(hosts=[f"http://{ip}:{port}"])
        logging.info(
            f"向Elasticsearch的{es_index}发起查询请求：\n{data2beautify(body)}"
        )

        query = es.search(
            index=es_index, body=body, _source=fields, scroll=scroll, size=10000
        )
        results = query["hits"]["hits"]
        total_count = query["hits"].get("total", 0)

        if isinstance(total_count, dict):
            total_count = total_count.get("value", 0)

        if total_count > 0:
            scroll_id = query["_scroll_id"]
            scroll_num = math.ceil(total_count / 10000)
            for _ in range(scroll_num):
                query_scroll = es.scroll(scroll_id=scroll_id, scroll=scroll)["hits"][
                    "hits"
                ]
                results.extend(query_scroll)

            for result in results:
                source_data = result.get("_source")
                if source_data is not None:
                    data.append(source_data)

    except (ConnectionError, TransportError) as e:
        logging.error(f"Elasticsearch 连接或查询失败: {str(e)}", exc_info=True)
    except Exception as e:
        logging.error(f"未知错误: {str(e)}", exc_info=True)
    finally:
        if es and scroll_id:
            try:
                es.clear_scroll(scroll_id=scroll_id)
            except Exception as e:
                logging.warning(f"清除 scroll_id 时出错: {str(e)}")

    end_time = time.time()
    take_time = end_time - start_time
    logging.info(f"查询Elasticsearch的{es_index}耗时{take_time:.2f}秒。")
    return data


def date_slice(start_time="2024-03-01 01:00:00", end_time="2024-03-01 23:00:00"):
    """
    获取两个日期时间之间每一天的开始时间和结束时间
    :param start_time: 开始日期时间
    :param end_time: 结束日期时间
    :return: 二维列表
    """
    slice_result = list()
    # 转换为datetime对象
    start_time_strp = datetime.datetime.strptime(start_time, "%Y-%m-%d %H:%M:%S")
    start_date = start_time_strp.strftime("%Y-%m-%d")
    end_time_strp = datetime.datetime.strptime(end_time, "%Y-%m-%d %H:%M:%S")
    end_date = end_time_strp.strftime("%Y-%m-%d")
    # 如果开始时间和结束时间在同一天，则直接返回切片
    if start_date == end_date:
        slice_result.append([start_time, end_time])
    else:
        # 生成日期列表
        date_list = []
        current_date = start_time_strp
        while current_date <= end_time_strp:
            date_list.append(current_date.strftime("%Y-%m-%d"))
            current_date += datetime.timedelta(days=1)
        # 输出每天的开始时间和截止时间
        for date in date_list:
            if date == start_date:
                date_start_time = start_time
                date_end_time = f"{date} 23:59:59"
            elif date == end_date:
                date_start_time = f"{date} 00:00:00"
                date_end_time = end_time
            else:
                date_start_time = f"{date} 00:00:00"
                date_end_time = f"{date} 23:59:59"
            slice_result.append([date_start_time, date_end_time])
    return slice_result


if __name__ == "__main__":
    print(get_nearby_day("2024-11-15", -30))
